
#include "mongoose.h"
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
// #include <czmq.h>
#include <zmq.h>
#include <unistd.h>
#include <signal.h>

#include <stdatomic.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>

#define QUEUE_SIZE 10240

typedef struct {
    char* data[QUEUE_SIZE];
    atomic_int head;
    atomic_int tail;
} LockFreeQueue;


void initQueue(LockFreeQueue *q) {
    atomic_init(&q->head, 0);
    atomic_init(&q->tail, 0);
}

int enqueue(LockFreeQueue *q, const char* value) {
    int tail = atomic_load(&q->tail);
    int next_tail = (tail + 1) % QUEUE_SIZE;

    if (next_tail == atomic_load(&q->head)) {
        // 队列已满
        return -1;
    }

    q->data[tail] = strdup(value);  // 分配并复制字符串
    atomic_store(&q->tail, next_tail);
    return 0;
}


int dequeue(LockFreeQueue *q, char** value) {
    int head = atomic_load(&q->head);

    if (head == atomic_load(&q->tail)) {
        // 队列为空
        return -1;
    }

    *value = q->data[head];
    atomic_store(&q->head, (head + 1) % QUEUE_SIZE);
    return 0;
}

LockFreeQueue q_ffr,q_web;   //传递的内容

static const char *s_listen_on = "ws://192.168.2.225:8000";
static const char *s_web_root = "./DMS_WEB";


static void start_thread(void *(*f)(void *), void *p) {
    #define closesocket(x) close(x)
    pthread_t thread_id = (pthread_t) 0;
    pthread_attr_t attr;
    (void) pthread_attr_init(&attr);
    (void) pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
    pthread_create(&thread_id, &attr, f, p);
    pthread_attr_destroy(&attr);
}

struct thread_data {
  struct mg_mgr *mgr;
  unsigned long conn_id;  // Parent connection ID
};

static void *thd_puller(void *param) {
  struct thread_data *p = (struct thread_data *) param;
  void *context = zmq_ctx_new();
  void *puller = zmq_socket(context, ZMQ_PULL);
  int rc = zmq_connect(puller, "tcp://127.0.0.1:5555");
  assert(rc == 0);
  for (;;) {
      char buffer[1024];
      memset(buffer,0,1024);
      zmq_recv(puller, buffer, 1024, 0);
      printf("Received:%s\n", buffer);
      mg_wakeup(p->mgr, p->conn_id, buffer, strlen(buffer));  // Send to parent
  }
  free(p);
  return NULL;
}



static void *thd_mg_task(void *param) {
  struct thread_data *p = (struct thread_data *) param;
  void *ctx_pull = zmq_ctx_new();
  void *puller = zmq_socket(ctx_pull, ZMQ_PULL);
  int rc; 
  rc = zmq_connect(puller, "tcp://127.0.0.1:5555");
  assert(rc == 0);

  void *ctx_req = zmq_ctx_new();
  void *req = zmq_socket(ctx_req, ZMQ_REQ);
  rc = zmq_connect(req, "tcp://127.0.0.1:5556");
  assert(rc == -1);
 
  char buffer[1024];
  for (;;) {
    char* web_cfg;
    usleep(10000);
    if (dequeue(&q_web, &web_cfg)==0)  // 有数据，发给所有client
    {
        memset(buffer,0,1024);
        printf("web send:%s\n",web_cfg);
        zmq_send(req , web_cfg, strlen(web_cfg), 0);
        zmq_recv(req, buffer,1024,0);
        printf("web recv:%s\n",buffer); 
        mg_wakeup(p->mgr, p->conn_id, buffer, strlen(buffer));
    }
    memset(buffer,0,1024);
    rc = zmq_recv(puller, buffer, 1024, ZMQ_DONTWAIT);
    if (rc != -1) {
        buffer[rc] = '\0';
        // printf("puller rx: %s\n",buffer);
        mg_wakeup(p->mgr, p->conn_id, buffer, strlen(buffer));  // Send to parent
    } else {
        if (errno==EAGAIN){
        }else{
            printf("unknown rc:%d\n",rc);
        }
    }
  }
  free(p);
  zmq_close(puller);
  return NULL;
}



// This RESTful server implements the following endpoints:
//   /websocket - upgrade to Websocket, and implement websocket echo server
//   /rest - respond with JSON string {"result": 123}
//   any other URI serves static files from s_web_root
static void fn(struct mg_connection *c, int ev, void *ev_data)
{
    if (ev == MG_EV_OPEN && c->is_listening){
        // c->is_hexdumping = 1;
        // Start worker thread
        printf("ev_open: c->id: %d\n",c->id);
        struct thread_data *data = calloc(1, sizeof(*data));  // Worker owns it
        data->conn_id = c->id;
        data->mgr = c->mgr;
        // start_thread(thd_puller, data);  // Start thread and pass data
        start_thread(thd_mg_task, data);
    }else if ( ev == MG_EV_WS_OPEN){
        printf("ec_connect c->id:%d\n",c->id);
        const char *json_req = "{\"type\":\"config\",\"data\":\"\"}";
        while(enqueue(&q_web,json_req)!=0){   // 只是发出去，并不等待结果，如何mg_ws_send给特定的？
            usleep(10);
            printf("ok");
        }
    }else if (ev == MG_EV_HTTP_MSG){
        struct mg_http_message *hm = (struct mg_http_message *)ev_data;
        if (mg_match(hm->uri, mg_str("/websocket"), NULL))
        {
            // Upgrade to websocket. From now on, a connection is a full-duplex
            // Websocket connection, which will receive MG_EV_WS_MSG events.
            mg_ws_upgrade(c, hm, NULL);
            c->data[0] = 'W';  // 设置一个标记，用来区分
        }
        else if (mg_match(hm->uri, mg_str("/rest"), NULL))
        {
            // Serve REST response
            mg_http_reply(c, 200, "", "{\"result\": %d}\n", 123);
        }
        else
        {
            // Serve static files
            struct mg_http_serve_opts opts = {.root_dir = s_web_root};
            mg_http_serve_dir(c, ev_data, &opts);
        }
    }else if (ev == MG_EV_WS_MSG){
        // Got websocket frame. Received data is wm->data. Echo it back!
        struct mg_ws_message *wm = (struct mg_ws_message *)ev_data;   // 收到信息
        // mg_ws_send(c, wm->data.buf, wm->data.len, WEBSOCKET_OP_TEXT);
        char buf[1024];
        memcpy(buf,wm->data.buf,wm->data.len);
        buf[wm->data.len]=0;
        printf("rx:%s\n",buf);
        // 此处会有不同的数据类型，读配置，更新配置，通过
        char sendbuf[2048];
        sprintf(sendbuf,"{\"type\":\"config\",\"data\":%s}",buf);
        while(enqueue(&q_web,sendbuf)!=0){   // 只是发出去，并不等待结果，如何mg_ws_send给特定的？
            usleep(10);
        }
        mg_iobuf_del(&c->recv, 0, c->recv.len);
    } else if (ev == MG_EV_WAKEUP) {
        struct mg_str *data = (struct mg_str *) ev_data;
        for (struct mg_connection *wc = c->mgr->conns; wc != NULL; wc = wc->next) {
            if (wc->data[0] == 'W') mg_ws_send(wc, data->buf, data->len, WEBSOCKET_OP_TEXT);
        }
    }

}




int main()
{
    int err;
    void *context = zmq_ctx_new();
    void *pusher = zmq_socket(context, ZMQ_PUSH);
    int rc = zmq_connect(pusher, "tcp://127.0.0.1:5556");
    assert(rc == 0);
    
    struct mg_mgr mgr; // Event manager
    mg_mgr_init(&mgr); // Initialise event manager
    printf("Starting WS listener on %s/websocket\n", s_listen_on);
    mg_http_listen(&mgr, s_listen_on, fn, NULL); // Create HTTP listener
    mg_wakeup_init(&mgr);
    for (;;){
        mg_mgr_poll(&mgr, 10); // Infinite event loop
        // char* web_cfg;
        // if (dequeue(&q_web, &web_cfg)==0)  // 有数据，发给所有client
        // {
        //     printf("rx web:%s\n",web_cfg);
        //     zmq_send(pusher,web_cfg,strlen(web_cfg),0);
        // }
    }
    mg_mgr_free(&mgr);
    return 0;
}



